package com.billionsfinance.hessian.common.zookeeper.client;


import com.billionsfinance.hessian.common.utils.ZkConfig;
import com.billionsfinance.hessian.model.ServerInfo;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Set;
import java.util.concurrent.CountDownLatch;

/**
 * @ FileName: ClientSideZkThread.java
 * @ Author: WeiHui-Z
 * @ Date: 2016-11-21  14:06
 * @ Version: v1.0.0
 */
public class ClientSideZkThread implements Runnable {

	private static final Logger LOGGER = LoggerFactory.getLogger(ClientSideZkThread.class);

	private ZkConfig zkConfig;

	private Set<ServerInfo> serverInfoSet;

	public ClientSideZkThread(ZkConfig zkConfig, Set<ServerInfo> serverInfoSet) {
		this.zkConfig = zkConfig;
		this.serverInfoSet = serverInfoSet;
	}

	@Override
	public void run() {
		String name = Thread.currentThread().getName();
		LOGGER.info("current running thread :" + name);

		RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, Integer.MAX_VALUE);

		CuratorFramework curator = CuratorFrameworkFactory.builder() //
				.connectString(zkConfig.getConnectString()) //  连接zookeeper的地址
				.sessionTimeoutMs(zkConfig.getSessionTimeoutMs())  //session 超时时间
				.connectionTimeoutMs(zkConfig.getConnectionTimeoutMs()) // 连接超时时间
				.retryPolicy(retryPolicy)  //重试策略
				.build();

		final CountDownLatch latch = new CountDownLatch(1);

		curator.getConnectionStateListenable().addListener(new ConnectionStateListener() {
			@Override
			public void stateChanged(CuratorFramework client, ConnectionState newState) {
				//连接之后，使计数减一
				if (newState == ConnectionState.CONNECTED) {
					latch.countDown();
					LOGGER.info("连接 zookeeper 成功!");
				}
			}
		});

		curator.start();
		try {
			//阻塞等待
			latch.await();
			//建立连接后，再建立路径监听器
			PathChildrenCacheListener pathChildrenCacheListener = new ClientSideZkPathChildrenCacheListener();
			// 绑定路径监听器
			for (ServerInfo serverInfo : serverInfoSet) {
				// 注册监听
				PathChildrenCache cache = new PathChildrenCache(curator, serverInfo.getPath(), true);
				cache.getListenable().addListener(pathChildrenCacheListener);
				cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}
